package com.yun.storm.Sales;

import java.math.BigDecimal;
import java.util.HashMap;
import java.util.Map;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

/**
 * 获取SalesSpout发送的数据，SalesTopology开启多线程。 给出每个线程处理的Sales数
 *
 * 在多线程情况下，对Sales数据只能局部汇总，不能整体汇总，可以把局部汇总的结果给一个单线程的BOLT进行整体汇总（SalesSumBolt）
 *
 * @author WangXin
 * @version V1.0
 */
public class SalesBolt extends BaseRichBolt {

	/**
	 *
	 */
	private static final long serialVersionUID = 1L;
	private OutputCollector collector;
	private Map<String, Integer> _countsSales = new HashMap<String, Integer>();
	private Map<String, Double> _countsAmount = new HashMap<String, Double>();

	/**
	 * 实例初始化的时候调用一次
	 *
	 * @param stormConf
	 *            The Storm configuration for this bolt.
	 * @param context
	 *            This object can be used to get information about this task's
	 *            place within the topology, including the task id and component
	 *            id of this task, input and output information, etc.
	 * @param collector
	 *            The collector is used to emit tuples from this bolt
	 */
	@Override
	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
		this.collector = collector;
	}

	/**
	 * Process a single tuple of input.
	 *
	 * @param input
	 *            The input tuple to be processed.
	 */
	@Override
	public void execute(Tuple input) {

		try {
			boolean insertFlg = false;
			int count = 0;
			double countAmount = 0;
			
			byte[] binaryByField = input.getBinaryByField("bytes");
	        String line = new String(binaryByField);		
	        
	        System.out.println("^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^");
	        System.out.println("line:"+line);
			
			String[] splitLog = line.split("\"");
//			String[] logFirst = line.split(", ");
//			String[] logFirst = line.split("\"");
			
			String productID = "";//商品ID
			String productName = "";//商品名称
			double OrderAmount = 0;//订单金额
			int productCount = 0;//商品数量
			String orderDate = "";//订单提交时间
			
//			for (String str1 : logFirst) {
//				splitLog = str1.split("\"");
//				if ("value = {".equals(splitLog[0])) {
			if (splitLog.length > 38) {
				if ("t_orderdetail".equals(splitLog[7]) && "insert".equals(splitLog[11])) {
					System.out.println("商品ID : " + splitLog[26].substring(1, splitLog[26].length() - 1));
					productID = splitLog[26].substring(1, splitLog[26].length() - 1);
					
					System.out.println("商品名称 : " + splitLog[33]);
					productName = splitLog[33];
					
					System.out.println("订单金额 : " + splitLog[38].substring(1, splitLog[38].length() - 1));
					OrderAmount = Double.valueOf(splitLog[38].substring(1, splitLog[38].length() - 1));
					
					System.out.println("商品数量 : " + splitLog[30].substring(1, splitLog[30].length() - 1));
					productCount = Integer.valueOf(splitLog[30].substring(1, splitLog[30].length() - 1));
					
					//统计商品销售量
					if (_countsSales.containsKey(productID)) {
						count = _countsSales.get(productID);
					} else {
						count = 0;
						insertFlg = true;
					}
					count = count + productCount;
					_countsSales.put(productID, count);
					
					//统计商品销售金额
					if (_countsAmount.containsKey(productID)) {
						countAmount = _countsAmount.get(productID);
					} else {
						countAmount = 0;
						insertFlg = true;
					}
					countAmount = countAmount+OrderAmount;
					_countsAmount.put(productID, countAmount);
					
					
					collector.emit(new Values(productID, productName, count, countAmount, orderDate, insertFlg));
					collector.ack(input);
					
				}
				if ("t_order".equals(splitLog[7]) && "insert".equals(splitLog[11])) {
					System.out.println("订单提交时间 : " + splitLog[35]);
					orderDate = splitLog[35];
					System.out.println("***********************");
				}
			}
					
					
					
//				}
//			}

		} catch (Exception e) {
			e.printStackTrace();
			this.collector.fail(input);
		}
	}

	/**
	 * Declare the output schema for all the streams of this topology.
	 *
	 * @param declarer
	 *            this is used to declare output stream ids, output fields, and
	 *            whether or not each output stream is a direct stream
	 */
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("productID", "productName", "count", "countAmount", "orderDate", "insertFlg"));
	}
	
}